package com.shujia.spark.core

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {
    /**
     * 1、创建spark环境，SparkContext
     */

    //spark配置文件对象
    val conf = new SparkConf()

    //指定运行模式，local：本地运行
    conf.setMaster("local")

    //spark程序名称
    conf.setAppName("wc")

    //创建spark运行环境对象
    val sc = new SparkContext(conf)

    /**
     * 2、读取文件
     *
     * RDD: 弹性的分布式数据集，类似一个集合
     */
    val linesRDD: RDD[String] = sc.textFile("data/words.txt")

    /**
     * 3、将一行转换成多行
     */
    val wordsRDD: RDD[String] = linesRDD.flatMap(line => line.split(","))

    /**
     * 4、转换成kv格式
     */
    val kvRDD: RDD[(String, Int)] = wordsRDD.map(word => (word, 1))

    /**
     * 5、统计单词的数量
     * reduceByKey: 按照key对value做聚合计算
     * (x, y) => x + y： 对valye做加和计算
     */
    val countRDD: RDD[(String, Int)] = kvRDD.reduceByKey((x, y) => x + y)


    //整理数据格式
    val resultRDD: RDD[String] = countRDD.map {
      case (word: String, count: Int) =>
        s"$word\t$count"
    }

    //删除输出目录
    val configuration = new Configuration()
    val fileSystem: FileSystem = FileSystem.newInstance(configuration)

    //判断是否存在，如果存在再删除
    //使用hdfsapi删除以存在的目录
    if (fileSystem.exists(new Path("data/word_count"))) {
      fileSystem.delete(new Path("data/word_count"), true)
    }

    /**
     * 6、保存结果
     */
    resultRDD.saveAsTextFile("data/word_count")


    //简写
    /*    sc.textFile("data/words.txt")
          .flatMap(_.split(","))
          .map((_, 1))
          .reduceByKey(_ + _)
          .foreach(println)*/
  }
}
